package com.zc.im.connector.listener;

import cn.hutool.core.collection.CollUtil;
import com.zc.im.common.config.ImConfigProperties;
import com.zc.im.common.constant.enums.EquipmentTypeEnum;
import com.zc.im.common.constant.enums.UserActionEnum;
import com.zc.im.common.core.domain.BaseUserInfo;
import com.zc.im.common.mq.dto.UserActionDto;
import com.zc.im.common.util.IpUtil;
import com.zc.im.connector.script.LuaScript;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Project：im
 * Date：2022/4/22
 * Time：14:44
 * Description：channel上下文
 *
 * @author Challen.Zhang
 * @version 1.0
 */
@Component
@RequiredArgsConstructor
public class ChannelContext {

    private final StringRedisTemplate redisTemplate;

    private final ImConfigProperties properties;

    private final RocketMQTemplate rocketMQTemplate;

    private HashOperations<String, EquipmentTypeEnum, String> hashOperations;

    @PostConstruct
    public void init() {
        hashOperations = redisTemplate.opsForHash();
    }

    private final Map<BaseUserInfo, Channel> channelMap = new ConcurrentHashMap<>();

    private final Map<ChannelId, BaseUserInfo> map = new ConcurrentHashMap<>();

    /**
     * 连接成功注册
     *
     * @param userInfo 用户基础信息
     * @param channel  channel
     */
    public void register(BaseUserInfo userInfo, Channel channel) {
        Channel persistentChannel = channelMap.get(userInfo);
        if (persistentChannel != null && !persistentChannel.id().asLongText().equals(channel.id().asLongText())) {
            //下线
            channelMap.put(userInfo, channel);
            map.remove(persistentChannel.id());
            map.put(channel.id(), userInfo);
            persistentChannel.writeAndFlush(UserActionDto.builder().userAction(UserActionEnum.OFFLINE).userInfo(userInfo).build());
            persistentChannel.close();
            return;
        }
        if (persistentChannel == null) {
            List<String> list = CollUtil.list(false, properties.getConnectPrefix() + userInfo.getUserId(), userInfo.getEquipmentType().toString(), IpUtil.getAddressAndPort());
            String addressAndPort = redisTemplate.execute(LuaScript.getReplaceConnector(), list);
            if (addressAndPort != null) {
                Message<UserActionDto> message = MessageBuilder.withPayload(UserActionDto.builder().userInfo(userInfo).userAction(UserActionEnum.OFFLINE).build()).build();
                rocketMQTemplate.syncSend(properties.getMqConfig().getActionTopic() + ":" + addressAndPort, message);
            }
            channelMap.put(userInfo, channel);
            map.put(channel.id(), userInfo);
        }
    }


    public void remove(ChannelId channelid) {
        offline(getUserInfo(channelid), false);
    }

    public BaseUserInfo getUserInfo(ChannelId channelId) {
        return map.get(channelId);
    }

    /**
     * 下线
     *
     * @param userInfo 用户信息
     * @param flag     是否发送下线消息
     */
    public void offline(BaseUserInfo userInfo, boolean flag) {
        Channel channel = channelMap.get(userInfo);
        if (channel != null) {
            hashOperations.delete(properties.getConnectPrefix() + userInfo.getUserId(), userInfo.getEquipmentType());
            channelMap.remove(userInfo);
            map.remove(channel.id());
            // 发送下线消息
            if (flag) {
                channel.writeAndFlush(UserActionDto.builder().userAction(UserActionEnum.OFFLINE).userInfo(userInfo).build());
            }
            channel.close();
        }

    }

}
